package com.suxin.module.sender.consumer;

import com.suxin.module.MyOpenTSDBConfig;
import com.suxin.module.bean.MyApi;
import com.suxin.module.bean.MyPoint;
import com.suxin.module.common.Json;
import com.suxin.module.http.MyHttpClient;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @Classname ConsumerRunnable
 * @Description [ 消费者任务 ]
 * @Author Tang
 * @Date 2020/10/16 11:23
 * @Created by ASUS
 */
@Slf4j
public class MyConsumerRunnable implements Runnable {

    private BlockingQueue<MyPoint> blockingQueue;

    private MyHttpClient myHttpClient;

    private MyOpenTSDBConfig myOpenTSDBConfig;

    private CountDownLatch countDownLatch;
    /**
     * 每次提交的大小
     */
    private int batchPutSize;

    /**
     * 每次提交等待的时间间隔，单位ms
     */
    private int batchPutTimeLimit;

    public MyConsumerRunnable(BlockingQueue<MyPoint> blockingQueue,
                              MyHttpClient myHttpClient,
                              MyOpenTSDBConfig myOpenTSDBConfig,
                              CountDownLatch countDownLatch) {
        this.blockingQueue = blockingQueue;
        this.myHttpClient = myHttpClient;
        this.myOpenTSDBConfig = myOpenTSDBConfig;
        this.countDownLatch = countDownLatch;
        this.batchPutTimeLimit = myOpenTSDBConfig.getBatchPutTimeLimit();
        this.batchPutSize = myOpenTSDBConfig.getBatchPutSize();
    }

    @Override
    public void run() {
        log.debug("thread:{} has started take point from queue", Thread.currentThread()
                .getName());
        // 等待取出的那个节点
        MyPoint waitPoint = null;
        boolean readyClose = false;

        int waitTimeLimit = batchPutTimeLimit / 3;

        while (!readyClose) {

            List<MyPoint> myPointList = new ArrayList<>();

            if (waitPoint != null) {
                myPointList.add(waitPoint);
            }
            // 循环从队列里面取数据，放到待发送集合里面
            for (int i = myPointList.size(); i < this.batchPutSize; i++) {
                try {
                    waitPoint = blockingQueue.poll(
                            waitTimeLimit, TimeUnit.MICROSECONDS
                    );
                    if (waitPoint != null) {
                        myPointList.add(waitPoint);
                    }
                } catch (InterruptedException e) {
                    readyClose = true;
                    log.info("The Thread {} is interrupted",Thread.currentThread()
                            .getName());
                }
            }
            // 如果列表为空，阻塞等待任务入队
            if (myPointList.size() == 0 && !readyClose) {
                try {
                    // 阻塞等待获取，如果线程池池被强制关闭了，就会触发中断异常
                    waitPoint = blockingQueue.take();
                } catch (InterruptedException e) {
                    // 异常发生时，代表线程池被关闭了
                    readyClose = true;
                    log.info("The Thread {} is interrupted",Thread.currentThread()
                            .getName());
                }
                continue;
            }

            if (myPointList.size() == 0) {
                continue;
            }
            setHttp(myPointList);
        }
        // 消费者-1
        this.countDownLatch.countDown();
    }

    /**
     * 发送请求写入数据
     * @param myPointList 数据点列表
     */
    private void setHttp(List<MyPoint> myPointList) {

        this.myHttpClient.post(
                MyApi.PUT.getPath(),
                Json.writeValueAsString(myPointList),
                null
        );

    }
}